"""
Case Type   : Ustore_Standby_Logicreplication
Case Name   : 开启复制槽时指定位置startPosition
Create At   : 2022/03/30
Owner       : zhao-pengcheng-g
Description :
    1.备份postgresql.conf,pg_hba.conf文件
    2.修改ustore参数
    3.修改逻辑复制相关参数
    4.创建数据库，创建用户，配置白名单
    5.jdbc准备
    6.使用jdbc创建复制槽,
    7.开启复制槽时指定为创建之前的位置
    8.开启复制槽时指定为此时此刻的位置
    9.开启复制槽时指定一个还未达到的位置
    10.开启复制槽时指定为创建之时的复制槽位置
    11.开启复制槽时指定为此刻之前，复制槽创建之后的位置
    12.删除复制槽
    13.资源清理，恢复参数
Expect      :
    1.备份成功
    2.修改ustore参数成功
    3.修改逻辑复制相关参数成功
    4.创建成功，配置白名单成功
    5.jdbc准备成功
    6.创建成功,
    7.开启成功,解码指定位置之后的sql语句
    8.开启成功,解码指定位置之后的sql语句
    9.开启成功,不解码任何sql语句
    10.开启成功,解码指定位置之后的sql语句
    11.开启成功,解码指定位置之后的sql语句
    12.删除逻辑复制槽成功
    13.资源清理成功，恢复参数成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

Primary_SH = CommonSH('PrimaryDbUser')

@unittest.skipIf(2 != Primary_SH.get_node_num(),'非1+1环境不执行')
class StandbyLogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.primary_root = Node('PrimaryRoot')
        self.sh_primy = CommonSH('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.dir_replication = os.path.join(macro.DB_BACKUP_PATH,
                                            'dir_replication_0022')
        self.postgresql_path = os.path.join(macro.DB_INSTANCE_PATH,
                                            macro.DB_PG_CONFIG_NAME)
        self.postgresql_bak_path = os.path.join(self.dir_replication,
                                                'postgresql_bak.conf')
        self.pg_hba_path = os.path.join(macro.DB_INSTANCE_PATH,
                                        macro.PG_HBA_FILE_NAME)
        self.pg_hba_bak_path = os.path.join(self.dir_replication,
                                            'pg_hba_bak.conf')
        self.db_name = 'db_replication_0022'
        self.user_name = 'u_logic_replication_0022'
        self.table_name = 't_standby_logic_replication_0022'
        self.slot_name = 'slot_standby_logic_replication_0022'
        self.jdbc_path = os.path.join(self.dir_replication, 'libs')
        self.CLASSPATH_PATH = os.path.join(self.dir_replication, 'CLASSPATH')
        self.jdbc_log_path = os.path.join(self.dir_replication, 'jdbc.log')
        self.log_file_path = os.path.join(self.dir_replication,
                                          'standby_logic_replication_0022.log')

        text = "-----step1:备份postgresql.conf,pg_hba.conf;expect:成功-----"
        self.logger.info(text)
        cp_cmd = f'''rm -rf {self.dir_replication} && 
                    mkdir {self.dir_replication} &&
                    cp {self.postgresql_path} {self.postgresql_bak_path} && 
                    cp {self.pg_hba_path} {self.pg_hba_bak_path} && echo OK'''
        cp_pri_msg = self.primary_node.sh(cp_cmd).result()
        self.logger.info(cp_pri_msg)
        self.assertEqual("OK", cp_pri_msg, "执行失败" + text)

        text = "-----step2:修改ustore参数;expect:成功-----"
        self.logger.info(text)
        self.config_ustore = 'enable_default_ustore_table'
        self.show_ustore_cmd = f'show {self.config_ustore};'
        check_res = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(check_res)
        self.ustore_default = check_res.splitlines()[-2].strip()
        if 'on' != self.ustore_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_ustore}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_ustore_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_ustore, 'on',
                                  '(1 row)', flag='1')

        text = "-----step3:修改逻辑复制相关参数;expect:成功-----"
        self.logger.info(text)
        self.config_slot_log = 'enable_slot_log'
        self.show_slot_log_cmd = f'show {self.config_slot_log};'
        check_res = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(check_res)
        self.slot_log_default = check_res.splitlines()[-2].strip()
        if 'on' != self.slot_log_default:
            guc_result = self.sh_primy.execute_gsguc('reload',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.sh_primy.execut_db_sql(self.show_slot_log_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_slot_log,
                                  'on', '(1 row)', flag='1')

        self.config_wal_level = 'wal_level'
        self.show_wal_level_cmd = f'show {self.config_wal_level};'
        check_res = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(check_res)
        self.wal_level_default = check_res.splitlines()[-2].strip()
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primy.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primy.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.sh_primy.execut_db_sql(self.show_wal_level_cmd)
        self.logger.info(show_msg)
        self.common.equal_sql_mdg(show_msg, self.config_wal_level,
                                  'logical', '(1 row)', flag='1')
        status = self.sh_primy.get_db_cluster_status()
        self.assertNotIn("ERROR", status, '执行失败:' + text)
        self.assertIn("Normal", status, '执行失败:' + text)

        text = "-----step4:创建数据库,创建用户,配置白名单;expect:成功-----"
        self.logger.info(text)
        create_cmd = f'''drop database if exists {self.db_name};
                    create database {self.db_name} template template0 \
                    encoding 'UTF8' lc_ctype 'en_US.UTF-8' \
                    lc_collate 'en_US.UTF-8';
                    drop user if exists {self.user_name};
                    create user {self.user_name} sysadmin \
                    password '{macro.COMMON_PASSWD}';'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname='postgres')
        self.logger.info(create_msg)
        self.assertIn(self.constant.CREATE_DATABASE_SUCCESS and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      create_msg, '执行失败' + text)
        param_1 = f'local replication {self.user_name}  sha256'
        param_2 = f'host replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host replication {self.user_name} ' \
            f'{self.primary_node.db_host}/32 sha256'
        result_1 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_1)
        result_2 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_2)
        result_3 = self.sh_primy.execute_gsguc('reload',
                                               self.constant.GSGUC_SUCCESS_MSG,
                                               '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step5:jdbc准备;expect:成功-----"
        self.logger.info(text)
        shell_cmd = f"mkdir {self.jdbc_path} && " \
            f"cd {macro.DB_SCRIPT_PATH}/../ && " \
            f"tar -xf `ls | grep -i jdbc | grep tar.gz ` -C {self.jdbc_path} " \
            f"&& chmod -R 777 {self.jdbc_path} && echo OK"
        self.logger.info(shell_cmd)
        shell_msg = self.primary_root.sh(shell_cmd).result()
        self.assertIn("OK", shell_msg, '执行失败' + text)

        self.logger.info("-----生成ClASSPATH-----")
        self.common.scp_file(self.primary_node, 'replication_tools.sh',
                             self.dir_replication)
        self.common.scp_file(self.primary_node, 'postgresql_right.jar',
                             self.dir_replication)
        self.tools_file_path = os.path.join(self.dir_replication,
                                            'replication_tools.sh')
        create_cmd = f"sh {self.tools_file_path} " \
            f"genClassPath {self.jdbc_path} > {self.CLASSPATH_PATH} && echo OK"
        self.logger.info(create_cmd)
        create_msg = self.primary_root.sh(create_cmd).result()
        self.assertIn("OK", create_msg, '执行失败' + text)

        self.logger.info("-----修改配置-----")
        self.common.scp_file(self.primary_node, 'conn.conf',
                             self.dir_replication)
        self.common.scp_file(self.primary_node, 'conn_rep1.conf',
                             self.dir_replication)
        self.common.scp_file(self.primary_node, 'conn_rep2.conf',
                             self.dir_replication)
        self.conf_files_path = os.path.join(self.dir_replication, 'conn*.conf')
        self.conf_1_path = os.path.join(self.dir_replication, 'conn_rep1.conf')
        self.conf_2_path = os.path.join(self.dir_replication, 'conn_rep2.conf')
        sed_cmd = f'''chmod 777 {self.dir_replication} && 
                    sed -i -e '/port=/c port={self.primary_node.db_port}' \
                    -e '/user=/c user={self.user_name}' \
                    -e "/password=/c password={macro.COMMON_PASSWD}" \
                    -e "/hostname=/c hostname={self.primary_node.db_host}" \
                    -e '/loggerfile=/c loggerfile={self.jdbc_log_path}' \
                    -e '/dbname=/c dbname={self.db_name}' \
                    -e '/loggerlevel=/c loggerlevel=error' \
                    {self.conf_files_path}'''
        self.logger.info(sed_cmd)
        sed_msg = self.primary_root.sh(sed_cmd).result()
        self.logger.info(sed_msg)
        self.assertNotIn("ERROR", sed_msg, '执行失败' + text)
        sed_cmd = f'''sed -i  -e \
                    "/hostname=/c hostname={self.standby_node.db_host}" \
                    -e '/port=/c port={self.standby_node.db_port}' \
                    {self.conf_2_path};'''
        self.logger.info(sed_cmd)
        sed_msg = self.primary_root.sh(sed_cmd).result()
        self.logger.info(sed_msg)
        self.assertNotIn("ERROR", sed_msg, '执行失败' + text)

    def test_standby_logicreplication(self):
        text = "-----step6:使用jdbc创建复制槽;expect:创建成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name};
            create table {self.table_name} 
            (id int primary key,lsn varchar, name varchar);
            insert into {self.table_name} values(1,'0/0000001', 'beforeall');'''
        create_msg = self.sh_primy.execut_db_sql(create_cmd,
                                                 dbname=self.db_name)
        self.logger.info(create_msg)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS and
                      self.constant.CREATE_TABLE_SUCCESS and
                      self.constant.INSERT_SUCCESS_MSG,
                      create_msg, '执行失败:' + text)
        self.common.scp_file(self.primary_node,
                             'Logical_replication_puls.class',
                             self.dir_replication)
        self.common.scp_file(self.primary_node,
                             'Logical_replication_puls.java',
                             self.dir_replication)
        create_cmd = f"source {macro.DB_ENV_PATH}; " \
            f"cd {self.dir_replication}; " \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_1_path} -M create_slot " \
            f"--args \"slotname={self.slot_name}\" "
        self.logger.info(create_cmd)
        create_msg = self.primary_node.sh(create_cmd).result()
        self.logger.info(create_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select count(*)||'c' from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name}';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        self.assertIn("1c", sql_msg, '执行失败:' + text)

        text = "-----step7:开启复制槽时指定为创建之前的位置;" \
               "expect:开启成功,解码指定位置之后的sql语句-----"
        self.logger.info(text)
        sql_cmd = f"select restart_lsn from pg_get_replication_slots() " \
            f"where slot_name='{self.slot_name}';" \
            f"insert into {self.table_name} values(2,'restart_lsn2','insert2');"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        restart_lsn = sql_msg.splitlines()[2].strip()
        before_lsn = hex(int(restart_lsn[2:], 16) - 5000)[2:]
        self.logger.info(before_lsn)
        lsn = '/'.join([restart_lsn[0], before_lsn]).upper()
        self.logger.info(lsn)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        start_cmd = f"cd {self.dir_replication};" \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_2_path} -M start_slot " \
            f"--args \"slotname={self.slot_name} " \
            f"startPosition={lsn} statusInterval=2 runtime=10\""
        self.logger.info(start_cmd)
        start_msg = self.primary_node.sh(start_cmd).result()
        self.logger.info(start_msg)
        self.assertIn(f'''"columns_val":["2","'restart_lsn2'","'insert2'"]''',
                      start_msg, '执行失败:' + text)

        text = "-----step8:开启复制槽时指定为此时此刻的位置;" \
               "expect:开启成功，解码指定位置之后的sql语句-----"
        self.logger.info(text)
        sql_cmd = f"insert into {self.table_name} values" \
            f"(3,'restart_lsn3','{self.slot_name}');" \
            f"select restart_lsn from pg_get_replication_slots() where " \
            f"slot_type='physical' order by restart_lsn desc limit 1;"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        lsn = sql_msg.splitlines()[-2].strip()
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        sql_cmd = f"insert into {self.table_name} values(4,'{lsn}','physical');"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        start_cmd = f"cd {self.dir_replication};" \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_2_path} -M start_slot " \
            f"--args \"slotname={self.slot_name} " \
            f"startPosition={lsn} statusInterval=2 runtime=10\""
        self.logger.info(start_cmd)
        start_msg = self.primary_node.sh(start_cmd).result()
        self.logger.info(start_msg)
        self.assertIn(f'''"columns_val":["4","'{lsn}'","'physical'"]''',
                      start_msg, '执行失败:' + text)

        text = "-----step9:开启复制槽时指定一个还未达到的位置;" \
               "expect:开启成功，不解码任何sql语句-----"
        self.logger.info(text)
        sql_cmd = f"insert into {self.table_name} " \
            f"values(5,'restart_lsn5','slot_name');" \
            f"select restart_lsn from pg_get_replication_slots() where " \
            f"slot_type='physical' order by restart_lsn desc limit 1;"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        current_lsn = sql_msg.splitlines()[-2].strip()
        after_lsn = hex(int(current_lsn[2:], 16) + 2000)[2:]
        self.logger.info(after_lsn)
        lsn = '/'.join([current_lsn[0], after_lsn]).upper()
        self.logger.info(lsn)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        start_cmd = f"cd {self.dir_replication};" \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_2_path} -M start_slot " \
            f"--args \"slotname={self.slot_name} " \
            f"startPosition={lsn} statusInterval=2 runtime=100\" " \
            f"> {self.log_file_path} & echo OK"
        self.logger.info(start_cmd)
        start_msg = self.primary_node.sh(start_cmd).result()
        self.logger.info(start_msg)
        text = "-----验证是否启动成功;expect:成功-----"
        self.logger.info(text)
        sql_cmd = f"select pg_sleep(5);" \
            f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name}';"
        thread_1 = ComThread(self.common.get_sh_result,
                             args=(self.primary_node, start_cmd,))
        thread_2 = ComThread(self.sh_standby.execut_db_sql,
                             args=(sql_cmd, '', self.db_name))
        thread_1.setDaemon(True)
        thread_2.setDaemon(True)
        thread_1.start()
        thread_2.start()
        thread_1.join(30)
        thread_2.join(30)
        msg_result_1 = thread_1.get_result()
        msg_result_2 = thread_2.get_result()
        self.logger.info(msg_result_1)
        self.logger.info(msg_result_2)
        self.assertEqual('t', msg_result_2.splitlines()[-2].strip(),
                         '执行失败:' + text)
        shell_cmd = f"cat {self.log_file_path}"
        shell_msg = self.primary_node.sh(shell_cmd).result()
        self.logger.info(shell_msg)
        self.assertEqual('', shell_msg, '执行失败:' + text)

        text = "-----step10:开启复制槽时指定为创建之时的复制槽位置;" \
               "expect:开启成功，解码指定位置之后的sql语句-----"
        self.logger.info(text)
        sql_cmd = f"insert into {self.table_name} " \
            f"values(6,'restart_lsn6','slot_name');" \
            f"select restart_lsn from pg_get_replication_slots() " \
            f"where slot_name='{self.slot_name}'"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        lsn = sql_msg.splitlines()[-2].strip()
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        start_cmd = f"cd {self.dir_replication};" \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_1_path} -M start_slot " \
            f"--args \"slotname={self.slot_name} " \
            f"startPosition={lsn} statusInterval=2 runtime=10\""
        self.logger.info(start_cmd)
        start_msg = self.primary_node.sh(start_cmd).result()
        self.logger.info(start_msg)
        self.assertIn(f'''"columns_val":["5","'restart_lsn5'","'slot_name'"]'''
                      and
                      f'''"columns_val":["6","'restart_lsn6'","'slot_name'"]''',
                      start_msg, '执行失败:' + text)

        text = "-----step11:开启复制槽时指定为此刻之前，复制槽创建之后的位置;" \
               "expect:开启成功,解码指定位置之后的sql语句-----"
        self.logger.info(text)
        sql_cmd = f"insert into {self.table_name} values" \
            f"(7,'restart_lsn7','{self.slot_name}');" \
            f"select restart_lsn from pg_get_replication_slots() where " \
            f"slot_type='physical' order by restart_lsn desc limit 1;" \
            f"select restart_lsn from pg_get_replication_slots() " \
            f"where slot_name='{self.slot_name}';"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        current_lsn = sql_msg.splitlines()[3].strip()
        mid_lsn = hex(int(current_lsn[2:], 16) - 1)[2:]
        self.logger.info(mid_lsn)
        lsn = '/'.join([current_lsn[0], mid_lsn]).upper()
        self.logger.info(lsn)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        sql_cmd = f"insert into {self.table_name} " \
            f"values(8,'{lsn}','mid_lsn');"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        start_cmd = f"cd {self.dir_replication};" \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_1_path} -M start_slot " \
            f"--args \"slotname={self.slot_name} " \
            f"startPosition={lsn} statusInterval=2 runtime=10\""
        self.logger.info(start_cmd)
        start_msg = self.primary_node.sh(start_cmd).result()
        self.logger.info(start_msg)
        self.assertIn(f'''"columns_val":["8","'{lsn}'","'mid_lsn'"]''',
                      start_msg, '执行失败:' + text)

    def tearDown(self):
        text = "-----step12:删除逻辑复制槽;expect:成功-----"
        self.logger.info(text)
        drop_slot_cmd = f"source {macro.DB_ENV_PATH};" \
            f"cd {self.dir_replication};" \
            f"java -cp `cat CLASSPATH` Logical_replication_puls " \
            f"-F {self.conf_1_path} -M drop_slot " \
            f"--args \"slotname={self.slot_name}\""
        self.logger.info(drop_slot_cmd)
        drop_slot_msg = self.primary_node.sh(drop_slot_cmd).result()
        self.logger.info(drop_slot_msg)
        self.logger.info("-----验证是否删除成功;expect:成功-----")
        sql_cmd = f"select * from pg_get_replication_slots();"
        sql_msg = self.sh_primy.execut_db_sql(sql_cmd, dbname=self.db_name)
        self.logger.info(sql_msg)

        text = "-----step13:资源清理，恢复参数，expect:成功-----"
        self.logger.info(text)
        text_1 = "-----step13.1:删除数据库，删除用户，expect:成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop database {self.db_name};
            drop user {self.user_name};'''
        drop_msg = self.sh_primy.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step13.2:恢复postgresql.conf,pg_hba.conf文件，" \
                 "expect:成功-----"
        self.logger.info(text_2)
        restore_cmd_1 = f'''
            if [[ -e {self.postgresql_path} && -e {self.postgresql_bak_path} ]];
            then rm -rf {self.postgresql_path};
            mv {self.postgresql_bak_path} {self.postgresql_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_1)
        restore_primary_1 = self.primary_node.sh(restore_cmd_1).result()
        self.logger.info(restore_primary_1)
        restore_cmd_2 = f'''
            if [[ -e {self.pg_hba_path} && -e {self.pg_hba_bak_path} ]];
            then rm -rf {self.pg_hba_path};
            mv {self.pg_hba_bak_path} {self.pg_hba_path};
            echo OK;
            else echo not exits; fi'''
        self.logger.info(restore_cmd_2)
        restore_primary_2 = self.primary_node.sh(restore_cmd_2).result()
        self.logger.info(restore_primary_2)

        text_3 = "-----step13.3:删除文件，expect:成功-----"
        self.logger.info(text_3)
        del_cmd = f'rm -rf {self.dir_replication};'
        self.logger.info(del_cmd)
        self.primary_node.sh(del_cmd)
        file_cmd = f'''if [ -d {self.dir_replication} ];
            then echo "exists"; else echo "not exists"; fi'''
        primy_file_msg = self.primary_node.sh(file_cmd).result()
        self.logger.info(primy_file_msg)
        self.assertNotIn(self.slot_name, sql_msg, '执行失败' + text)
        self.assertIn(self.constant.DROP_DATABASE_SUCCESS and
                      self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertEqual('OK', restore_primary_1, '执行失败' + text_2)
        self.assertEqual('OK', restore_primary_2, '执行失败' + text_2)
        self.assertEqual("not exists", primy_file_msg, '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")